package org.shj.hadoop.fs;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.MapFile;
import org.apache.hadoop.io.MapFile.Reader;
import org.apache.hadoop.io.MapFile.Writer;
import org.apache.hadoop.io.SequenceFile;
import org.apache.hadoop.io.SequenceFile.CompressionType;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.Writable;
import org.apache.hadoop.io.WritableComparable;
import org.apache.hadoop.util.ReflectionUtils;
import org.apache.zookeeper.common.IOUtils;

/**
 * MapFile 是排过序的 Sequence File,键必须是WritableComparable实例，值必须是Writable类型实例。
 * 生成的 MapFile 其实是目录，其中包含index和data文件。 
 * @author Administrator
 *
 */
public class MapFileDemo {

	public static final String[] DATA = {
			"One, two, buckle my shoe",
			"Three, four, shut the door",
			"Five ,six, pick up sticks",
			"Seven, eight, lay them straight",
			"nine ,ten, abig fat hen"
	};
	
	
	public static void writeFile() throws Exception{
		String path = "/user/shj/test1.map";
		Configuration cfg = new Configuration();
		cfg.set("fs.defaultFS", "hdfs://192.168.56.101:9000");
		
		Path p = new Path(path);
		
		IntWritable key = new IntWritable();
		Text value = new Text();
		MapFile.Writer writer = null;
		
		try{
			SequenceFile.Writer.Option opt1 = SequenceFile.Writer.file(p);
			MapFile.Writer.Option opt2 = Writer.keyClass(IntWritable.class);
			SequenceFile.Writer.Option opt3 = Writer.valueClass(Text.class);

			//这里有压缩选项，所以直接view 文件的话， 会看到乱码
			writer = new MapFile.Writer(cfg, p, opt1, opt2, opt3, Writer.compression(CompressionType.RECORD));
			
			for(int i = 0 ; i < 1024; i++){
				key.set(i + 1);
				value.set(DATA[ i % DATA.length]);
				writer.append(key, value);
			}
		}finally{
			IOUtils.closeStream(writer);
		}
	}
	
	public static void readFile() throws Exception{
		String path = "/user/shj/test.map";
		Configuration cfg = new Configuration();
		cfg.set("fs.defaultFS", "hdfs://192.168.56.101:9000");
				
		Path p = new Path(path);
		
		SequenceFile.Reader.Option opt1 = SequenceFile.Reader.file(p);
//		SequenceFile.Reader.Option opt2 = Reader.keyClass(IntWritable.class);
//		SequenceFile.Reader.Option opt3 = Reader.valueClass(Text.class);
				
		MapFile.Reader reader = null;
		try{
			//实例化reader 应该仿照上面writeFile()中方法。
			reader = new MapFile.Reader(p,cfg,opt1);
			
			WritableComparable<IntWritable> key = (WritableComparable<IntWritable>)ReflectionUtils.newInstance(reader.getKeyClass(), cfg) ;
			Writable value = (Writable)ReflectionUtils.newInstance(reader.getValueClass(), cfg);
			
			while(reader.next(key, value)){
				System.out.printf("[%s]\t%s\n", key, value);
			}
			
		}finally{
			IOUtils.closeStream(reader);
		}
	}
	
	public static void main(String[] args) throws Exception{
		writeFile();
//		readFile();
	}
}
